package com.boguan.web.controller.pulsar;

import org.apache.pulsar.client.api.*;

import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;

public class ConsumerExample {

    static String appId = "2510";
    static String appSecret = "1854c2255e2f4b3545c96c9c71b2cd03f5f04512ddcdbc601d5245d0355adce4";
    static String serviceUrl = "pulsar://pulsar.weiyuzn.com:6650";


    private PulsarClient client = null;
    private Consumer consumer = null;


    @SuppressWarnings("InfiniteLoopStatement")
    public static void main(String[] args) throws PulsarClientException, NoSuchAlgorithmException, InvalidKeyException {

        PulsarClient client = PulsarClient.builder()
                .serviceUrl(serviceUrl)
                .authentication(new WeiYuAuthentication(appId, appSecret))
                .build();

        Consumer<byte[]> consumer = client.newConsumer()
                .topic("persistent://" + appId + "/out/deviceState")
                .subscriptionName(appId)
                .subscriptionType(SubscriptionType.Key_Shared)
                .deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(3).build())
                .subscribe();

        do {

            // Wait for a message
            Message<byte[]> msg = consumer.receive();

            try {
                // Do something with the message
                String data=new String(msg.getData());
                System.out.println("Message received: " + data);

                // Acknowledge the message so that it can be deleted by the message broker
                consumer.acknowledge(msg);
            } catch (Exception e) {
                // Message failed to process, redeliver later
                consumer.negativeAcknowledge(msg);
            }
        } while (true);
    }



}
